#include "config.h"
#include <string.h>

#include "xIPC.h"
#include "aux_types.h"
#include "aux_errno.h"
#include "aux_socket.h"
#include "aux_string.h"
#include "aux_time.h"
#include "aux_ip.h"
#include "aux_messagemanager.h"
#include "aux_debugmsg.h"
#include "aux_message.h"
#include "aux_iointerface.h"
#include "aux_buffer.h"

static EErrNo s_xIPC_Pipe_MsgHandler(struct message_listener_t *ptMsgListener,
                                        TMessage *ptMsg)
{
    EErrNo eRet;
    TMessage tMsg;

    if (!ptMsgListener || !ptMsg)
        return -ERRNO_EPARAMETER_INVALID;

    if ((!((ptMsgListener->uOwner).ptClient)) || (!(ptMsgListener->ptPipe)))
        return -ERRNO_EPARAMETER_INVALID;

    eRet = xIPC_Msg_DeepCopy(&tMsg, ptMsg);
    if (ERRNO_SUCCESS != eRet)
        return ERRNO_FAIL;

    aux_prt_debug("pipe has one msg, client %s, sender %s, data len %d\n",
                  ((ptMsgListener->uOwner).ptClient)->szClient,
                  ptMsg->szSender, ptMsg->nDataByteCnt);
    eRet = xIPC_Msg_AddMessage(
                                    &(ptMsgListener->ptPipe->tMsgRecvManager),
                                    &tMsg);
    if (-ERRNO_EADDMSGFAIL == eRet)
    {
        aux_prt_err("pipe add to msg queue fail\n");
        if (tMsg.pData)
            free(tMsg.pData);
        tMsg.pData = NULL;
        tMsg.nDataByteCnt = 0;
        return ERRNO_FAIL;
    }
    return ERRNO_SUCCESS;
}

TxIPCIOPipe *xIPC_Pipe_Get(char *pszIp,
                                 u16   wPort,
                                 char *pszName,
                                 char *pszListenRemote)
{
#define __FUNCTION_NAME__ "xIPC_Pipe_Get"
    TxIPCIOPipe       *ptPipe = NULL;
    TMessageListener   tListener;

    if (!pszIp)
        pszIp = LOCAL_IP_STR;

    if (!pszName)
        return NULL;

    if (pszListenRemote && strlen(pszListenRemote) >= TERMINAL_NAME_LENGTH)
        return NULL;

    ptPipe = (TxIPCIOPipe *)malloc(sizeof(TxIPCIOPipe));
    if (!ptPipe)
        return NULL;

    memset(ptPipe, 0, sizeof(TxIPCIOPipe));

    ptPipe->eState = IPC_PIPE_UNKNOWN;

    if (ERRNO_SUCCESS !=
        aux_buf_init_buffer_with_lock(&(ptPipe->tDataBufLocked)))
    {
        free(ptPipe);
        return NULL;
    }
    if (ERRNO_SUCCESS != xIPC_Msg_Manager_Init(&(ptPipe->tMsgRecvManager),
                                        "pipe message buffer",
                                        "listener for pipe msg"))
    {
        aux_buf_destroy_buffer_with_lock(&(ptPipe->tDataBufLocked));
        free(ptPipe);
        return NULL;
    }

    if (ERRNO_SUCCESS != xIPC_Client_Init(&(ptPipe->tClient),
                                          pszIp, wPort, pszName))
    {
        xIPC_Msg_Deinit(&(ptPipe->tMsgRecvManager));
        aux_buf_destroy_buffer_with_lock(&(ptPipe->tDataBufLocked));
        free(ptPipe);
        return NULL;
    }

    memset(&tListener, 0, sizeof(TMessageListener));
    if (!pszListenRemote)
    {
        tListener.eListenFilterType = LISTEN_TYPE_ALL;
    }
    else
    {
        tListener.eListenFilterType = LISTEN_TYPE_NAME;
        strcpy(&((tListener.szSender)[0]), pszListenRemote);
    }

    tListener.uOwner.ptClient   = &(ptPipe->tClient);
    tListener.ptPipe            = ptPipe;
    tListener.pMessageHandler   = s_xIPC_Pipe_MsgHandler;
    if (ERRNO_SUCCESS !=
        xIPC_Client_AddListener(&(ptPipe->tClient), &tListener))
    {
        xIPC_Client_Deinit(&(ptPipe->tClient));

        xIPC_Msg_Deinit(&(ptPipe->tMsgRecvManager));
        aux_buf_destroy_buffer_with_lock(&(ptPipe->tDataBufLocked));
        free(ptPipe);
        return NULL;
    }

    ptPipe->eState = IPC_PIPE_BIDIRECTION;
    return ptPipe;
#undef __FUNCTION_NAME__
}

int xIPC_Pipe_Put(TxIPCIOPipe *ptPipe)
{
    if (!ptPipe || IPC_PIPE_UNKNOWN == ptPipe->eState)
        return 0;

    xIPC_Client_Deinit(&((ptPipe->tClient)));
    xIPC_Msg_Deinit(&(ptPipe->tMsgRecvManager));
    aux_buf_destroy_buffer_with_lock(&(ptPipe->tDataBufLocked));
    ptPipe->eState = IPC_PIPE_UNKNOWN;
    free(ptPipe);
    return 0;
}

int xIPC_Pipe_Send(TxIPCIOPipe *ptPipe, char *pszDest, u8 *pabyData, u32 nLen)
{
    TMessage tMsg;
    EErrNo   eRet;
    u8      *pabySend;
    u32      nPackageDataSize = 0;
    s32      nSizeLeftData = nLen;
    u32      nSizeAlreadySend = 0;
    s32      nSizeCurrentSend = nLen;

    if (!ptPipe || !pabyData || !pszDest ||
        (0 == nLen) || (TERMINAL_NAME_LENGTH <= strlen(pszDest)))
        return -ERRNO_EPARAMETER_INVALID;

    if (ptPipe->eState != IPC_PIPE_SEND &&
        ptPipe->eState != IPC_PIPE_BIDIRECTION)
        return -ERRNO_ESTATUSERROR;

    if (ERRNO_SUCCESS !=
        xIPC_Msg_GetDataMaxSize(&nPackageDataSize) ||
        0 == nPackageDataSize) /*each message can hold max size of data */
        return -ERRNO_EINTERNALERROR;

    memset(&tMsg, 0, sizeof(tMsg));
    tMsg.eMessageDestinationType = MESSAGE_DST_TYPE_TO_CLIENT;
    tMsg.eMessageId          = MESSAGE_ID_CLIENT_SEND_COMMON_MESSAGE_TO_ROUTER;
    tMsg.eReceiverFilterType = RECEIVER_FILTER_NAME;
    strcpy(tMsg.szReceiverFilter, pszDest);

    while (nSizeLeftData > 0)
    {
        nSizeCurrentSend = nPackageDataSize; /* send max default */
        if (nSizeCurrentSend > nSizeLeftData)
            nSizeCurrentSend = nSizeLeftData;

        pabySend = (u8 *) malloc(nSizeCurrentSend); /* will passed to sender */
        if (!pabySend)
            return nSizeAlreadySend;

        aux_memcpy_each_byte(pabySend,
                             &(pabyData[nSizeAlreadySend]),
                             nSizeCurrentSend);
        tMsg.pData = pabySend;
        tMsg.nDataByteCnt = nSizeCurrentSend;

        eRet = xIPC_Client_Send(&(ptPipe->tClient), &tMsg);

        if (ERRNO_SUCCESS == eRet || ERRNO_DELAY == eRet)
        {
            pabySend = NULL;
            nSizeLeftData -= nSizeCurrentSend;
            nSizeAlreadySend += nSizeCurrentSend;
            continue;
        }
        else if (-ERRNO_EADDMSGFAIL == eRet)
        {
            free(pabySend);
            return nSizeAlreadySend;
        }
        else
        {
            return nSizeAlreadySend;
        }
    }

    return nSizeAlreadySend;;
}

int xIPC_Pipe_GetSendingToRemoteStatus(TxIPCIOPipe *ptPipe)
{
    if (!ptPipe)
        return -ERRNO_EPARAMETER_INVALID;

    if (ptPipe->eState != IPC_PIPE_SEND &&
        ptPipe->eState != IPC_PIPE_BIDIRECTION)
        return ERRNO_INVALID;

    if (isThreadDead((ptPipe->tClient).eSenderProcessState))
        return ERRNO_INVALID;

    return ERRNO_VALID;
}

int xIPC_Pipe_GetRecvingFromRemoteStatus(TxIPCIOPipe *ptPipe)
{
    if (!ptPipe)
        return -ERRNO_EPARAMETER_INVALID;

    if (ptPipe->eState != IPC_PIPE_RECV &&
        ptPipe->eState != IPC_PIPE_BIDIRECTION)
        return ERRNO_INVALID;

    if (isThreadDead((ptPipe->tClient).eListenerProcessState))
        return ERRNO_INVALID;

    return ERRNO_VALID;
}

int xIPC_Pipe_Wait(TxIPCIOPipe *ptPipe, u8 *pabyBuffer,
                        u32 nWantLen, u32 nTimeoutMs)
{
    int nReadCnt = 0;
    u32 nTotalRead = 0;

    if (!ptPipe || !pabyBuffer || 0 == nTimeoutMs)
        return -ERRNO_EPARAMETER_INVALID;

    if (0 == nWantLen)
        return 0;

    if (ptPipe->eState == IPC_PIPE_UNKNOWN)
        return -ERRNO_ESTATUSERROR;

    nReadCnt = xIPC_Pipe_Read(ptPipe, pabyBuffer, nWantLen);
    if (nReadCnt > 0)
        nTotalRead = nReadCnt;

    if (nTotalRead >= nWantLen)
        return nTotalRead;

    while (nTotalRead < nWantLen)
    {
        if (ERRNO_SUCCESS != xIPC_Msg_WaitForMsgTimeout(
                        &(ptPipe->tMsgRecvManager), nTimeoutMs))
            return nTotalRead;
        nReadCnt = xIPC_Pipe_Read(ptPipe,
                                  &(pabyBuffer[nTotalRead]),
                                  nWantLen - nTotalRead);
        if (nReadCnt >= 0)
            nTotalRead += nReadCnt;
        else
            return nTotalRead;
    }

    return nTotalRead;
}

int xIPC_Pipe_Read(TxIPCIOPipe *ptPipe, u8 *pabyBuffer, u32 nWantLen)
{
#define __FUNCTION_NAME__ "xIPC_Pipe_Read"
    EErrNo    eRet;
    u32       nReadTotalCnt = nWantLen;
    TMessage *ptMsg = NULL;
    u32       nCopyCnt = 0;
    u8       *pabyData = NULL;
    u32       dataSize = 0;
    if (!ptPipe || !pabyBuffer)
        return -ERRNO_EPARAMETER_INVALID;

    if (0 == nWantLen)
        return 0;

    if (ptPipe->eState == IPC_PIPE_UNKNOWN)
        return -ERRNO_ESTATUSERROR;
    /* try to read from buffer */

    eRet = aux_buf_lock_read(&(ptPipe->tDataBufLocked),
                                   pabyBuffer, &nReadTotalCnt);
    aux_prt_debug("try read %d, get from buf %d now!\n", nWantLen, nReadTotalCnt);
    if (ERRNO_SUCCESS == eRet && nReadTotalCnt >= nWantLen)
    {
        aux_prt_debug("get %d bytes from buffer\n", nReadTotalCnt);
        return nReadTotalCnt;
    }
    /* buffer not enough */
    while (nReadTotalCnt < nWantLen)
    {
        eRet = xIPC_Msg_PickMsg(
                                        &(ptPipe->tMsgRecvManager), &ptMsg);
        if (eRet != ERRNO_SUCCESS)
            break;

        if (!ptMsg ||
            (0 == ptMsg->nDataByteCnt) ||
            (NULL == ptMsg->pData))
            continue;

        pabyData = (u8 *)(ptMsg->pData);
        dataSize = ptMsg->nDataByteCnt;
        aux_prt_debug("pop a node, data len %d\n", dataSize);
        if (nReadTotalCnt + dataSize > nWantLen)
        {
            nCopyCnt = nWantLen - nReadTotalCnt;
        }
        else
        {
            nCopyCnt = dataSize;
        }

        memcpy(&(pabyBuffer[nReadTotalCnt]), pabyData, nCopyCnt);
        nReadTotalCnt += nCopyCnt;

        if (nCopyCnt < dataSize)
        {
            /* put the data left from queue into the buffer */
            aux_buf_lock_append(&(ptPipe->tDataBufLocked),
                                      &(pabyData[nCopyCnt]),
                                      dataSize - nCopyCnt);
        }

        /* release the memory */
        free(ptMsg);
    }

    return nReadTotalCnt;
#undef __FUNCTION_NAME__
}




TxIPCMessageRouter *xIPC_Router_Get(char *pszIp, u16 wPort, char *pszName)
{
#define __FUNCTION_NAME__ "xIPC_Router_Get"
    TxIPCMessageRouter *ptRouter = NULL;

    if (!pszIp || !pszName)
        return NULL;

    ptRouter = (TxIPCMessageRouter *)malloc(sizeof(TxIPCMessageRouter));
    if (!ptRouter)
        return NULL;

    if (ERRNO_SUCCESS != xIPC_Router_Init(ptRouter, pszIp, wPort, pszName))
    {
        free(ptRouter);
        return NULL;
    }

    return ptRouter;
#undef __FUNCTION_NAME__
}

int xIPC_Router_Put(TxIPCMessageRouter *ptRouter)
{
#define __FUNCTION_NAME__ "xIPC_Router_Put"

    if (!ptRouter)
        return 0;

    if (ERRNO_SUCCESS != xIPC_Router_Deinit(ptRouter))
        return -1;

    free(ptRouter);
    return 0;
#undef __FUNCTION_NAME__
}


EErrNo xIPCCoreIOInterfaceInit(void)
{
    return ERRNO_SUCCESS;
}

EErrNo xIPCCoreIOInterfaceDeinit(void)
{
    return ERRNO_SUCCESS;
}


